package en.edu.zxj.mq.mqserver.core;

import en.edu.zxj.mq.common.Consumer;
import en.edu.zxj.mq.common.ConsumerEnv;
import en.edu.zxj.mq.common.MqException;
import en.edu.zxj.mq.mqserver.VirtualHost;
import lombok.extern.slf4j.Slf4j;

import java.util.List;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingQueue;

/**
 * Created with IntelliJ IDEA.
 * Description：
 *
 * @author: zxj
 * @date: 2024-03-02
 * @time: 12:24:15
 */
@Slf4j
public class ConsumerManager {
    // parent 用来记录虚拟主机
    private final VirtualHost parent;
    // 存放令牌的队列, 通过令牌来触发消费线程的消费操作
    // 使用一个阻塞队列来触发消息消费, 称为令牌队列, 每次有消息过来了, 都为队列中放一个令牌(也就是队列名), 让后消费者再去消费对应的队列信息
    // 作用: 令牌队列的设定, 避免搞出来太多线程, 否则就需要给每个队列都安排一个单独的线程了, 如果队列很多则开销就比较的了
    private final BlockingQueue<String> tokenQueue = new LinkedBlockingQueue<>();
    // 使用一个线程池用来执行消息的回调
    private final ExecutorService workerPool = Executors.newFixedThreadPool(4);

    // 扫描线程
    private Thread scannerThread = null;

    public ConsumerManager(VirtualHost virtualHost) {
        this.parent = virtualHost;

        scannerThread = new Thread(() -> {
            while (true) {
                try {
                    // 1. 拿到令牌
                    String queueName = tokenQueue.take();
                    // 2. 根据令牌找到对应的队列
                    MSGQueue msgQueue = parent.getMemoryDataCenter().getMSGQueue(queueName);
                    if (msgQueue == null) {
                        throw new MqException("获取令牌后, 发现队列为空! queueName: " + queueName);
                    }
                    // 3. 从队列中消费一次消息
                    consumeMessage(msgQueue);
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        });

        scannerThread.start();
    }

    /**
     * @description: 通知消费者去消费消息
     **/
    public void notifyConsume(String name) throws InterruptedException {
        tokenQueue.put(name);
    }


    /**
     * @description: 添加订阅者
     * 新来的消费者需要消费掉之前的消息
     **/
    public void addConsumer(String consumerTag, String queueName, boolean autoAck, Consumer consumer) throws MqException {
        // 找到对应的队列
        MSGQueue queue = parent.getMemoryDataCenter().getMSGQueue(queueName);
        if (queue == null) {
            throw new MqException("[ConsumerManager] 队列不存在, queueName: " + queueName);
        }
        ConsumerEnv consumerEnv = new ConsumerEnv(consumerTag, queueName, autoAck, consumer);

        synchronized (queue) {
            queue.addConsumerEnv(consumerEnv);
            // 如果当前队列中已经有了一些消息, 需要立即消费掉
            int n = parent.getMemoryDataCenter().getMessageCount(queueName);
            for (int i = 0; i < n; i++) {
                // 这个方法调用一次就消费一次消息
                consumeMessage(queue);
            }
        }
    }

    /**
     * @description: 消费一次消息
     **/
    private void consumeMessage(MSGQueue queue) {
        // 1. 按照轮询的方式, 取出消费者
        ConsumerEnv luckyDog = queue.chooseConsumer();
        if (luckyDog == null) {
            // 暂时还没有消费者, 就暂时不消费
            return;
        }
        // 2. 从指定队列中取出一个元素
        Message message = parent.getMemoryDataCenter().pollMessage(queue.getName());
        if (message == null) {
            return;
        }
        // 3. 把消息带入到消费者的回调方法中, 丢给线程池执行
        workerPool.submit(() -> {
            try {
                // 1. 把消息放到待确认的集合中, 这个操作势必在执行回调之前
                parent.getMemoryDataCenter().addMessageWaitAck(queue.getName(), message);
                // 2. 真正执行回调操作
                luckyDog.getConsumer().handleDelivery(luckyDog.getConsumerTag(), message.getBasicProperties(), message.getBody());
                // 3. 如果当前是 "自动应答", 就可以直接报消息给删除了
                //    如果当前是 "手动应答", 则先不处理, 交给后续消费调用 basicAck 方法来处理
                if (luckyDog.isAutoAck()) {
                    // ① 先删除磁盘上的消息
                    if (message.getDeliverMode() == 2) {
                        parent.getDiskDataCenter().deleteMessage(queue, message);
                    }
                    // ② 删除上面的待确认集合中的消息
                    parent.getMemoryDataCenter().removeMessageWaitAck(queue.getName(), message.getMessageId());
                    // ③ 删除内存中消息
                    parent.getMemoryDataCenter().deleteMessage(message.getMessageId());

                    log.info("消息被消费成功, queueName: {}", queue.getName());

                }
            } catch (Exception e) {
                e.printStackTrace();
            }
        });
    }


}
